package com.aiwiown.snackmq.broker.handler;

import com.aiwiown.snackmq.cluster.metadata.MetadataManager;
import com.aiwiown.snackmq.common.handler.MessageHandler;
import com.aiwiown.snackmq.common.handler.RequestContext;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.common.metrics.OffsetQueryable;
import lombok.extern.slf4j.Slf4j;

/**
 * 处理获取位移请求的处理器。
 * 该处理器负责处理来自客户端的获取位移请求，通过查询位移信息并返回给客户端。
 */
@Slf4j
public class FetchOffsetHandler implements MessageHandler {

    // 用于查询位移的接口
    private final OffsetQueryable offsetQueryable;
    // 用于管理集群元数据的管理器
    private final MetadataManager metadataManager;

    /**
     * 构造函数，初始化FetchOffsetHandler。
     *
     * @param offsetQueryable 位移查询接口实现
     * @param metadataManager 集群元数据管理器
     */
    public FetchOffsetHandler(OffsetQueryable offsetQueryable, MetadataManager metadataManager) {
        this.offsetQueryable = offsetQueryable;
        this.metadataManager = metadataManager;
    }

    /**
     * 处理获取位移的请求。
     * 该方法解析请求中的消费者组、主题和分区信息，然后查询相应的位移并返回给客户端。
     * 如果查询成功，返回成功响应；如果查询失败，返回错误响应。
     *
     * @param ctx     请求上下文
     * @param request 请求消息
     */
    @Override
    public void handle(RequestContext ctx, Message request) {
        try {
            // 【核心修复】: 使用 "consumerGroup" 作为属性键，与客户端协议保持一致。
            String consumerGroup = request.getProperty("consumerGroup");
            String topic = request.getTopic();
            String partitionStr = request.getProperty("partition");

            // 增强参数校验
            if (consumerGroup == null || topic == null || partitionStr == null) {
                throw new IllegalArgumentException("Fetch offset request must contain consumerGroup, topic, and partition properties.");
            }

            int partitionId = Integer.parseInt(partitionStr);

            // 从状态机（或任何实现了 OffsetQueryable 的组件）获取位移
            long offset = offsetQueryable.fetchOffset(consumerGroup, topic, partitionId);

            // 构建成功的响应
            Message response = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.SUCCESS)
                    .property("offset", String.valueOf(offset))
                    .build();
            ctx.writeResponse(response);

        } catch (Exception e) {
            log.error("Failed to fetch offset for request: {}", request, e);
            // 构建失败的响应
            Message errorResponse = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.FAILED)
                    .errorMessage(e.getMessage())
                    .build();
            ctx.writeResponse(errorResponse);
        }
    }

    /**
     * 获取该处理器处理的消息类型。
     * 该方法返回MessageType.FETCH_OFFSET，表示该处理器处理获取位移类型的请求。
     *
     * @return 消息类型
     */
    @Override
    public MessageType getMessageType() {
        return MessageType.FETCH_OFFSET;
    }
}
